package com.pad.dpscn.daag.sdk.core.async;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.function.BiConsumer;
import java.util.function.Function;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.pad.dpscn.daag.sdk.common.constant.OperationTypeModel;
import com.pad.dpscn.daag.sdk.common.constant.SqlConstants;
import com.pad.dpscn.daag.sdk.common.enums.ErrorCodeEnum;
import com.pad.dpscn.daag.sdk.common.exception.DaagException;
import com.pad.dpscn.daag.sdk.common.message.MqMessage;
import com.pad.dpscn.daag.sdk.common.utils.SpringUtils;
import com.pad.dpscn.daag.sdk.core.properties.DaagProperties;
import com.pad.dpscn.daag.sdk.core.strategy.service.MessageQueueService;
/**
 * 
 * com.pad.dpscn.daag.sdk.core.async  
 *  
 * <p>  
 * Description:用于异步查询数据库并发送消息  
 * </p>  
 *  
 * @author yangdezhao
 * @since 2024年3月11日  
 * @version 1.0
 */
public class AsyncQueryDbAndSendMsgServiceImpl implements AsyncQueryDbAndSendMsgService {

	private static final Log log = LogFactory.getLog(AsyncQueryDbAndSendMsgServiceImpl.class);

	@Autowired
	private SqlSessionFactory sqlSessionFactory;
	@Autowired
	private DaagProperties daagProperties;
	
	
	
	/**
	 * 异步查询数据库并发送消息。 
     * @param dataQueue 包含要处理消息的队列  
	 * @return void
	 */
	@Override
	public void asyncQueryDbAndSendMsg(ConcurrentLinkedQueue<MqMessage> dataQueue) {
		CompletableFuture.supplyAsync(() -> {
			log.info("=====================异步操作begin========================");
			log.info("asyncQueryDbAndSendMsg() execution begin: dataQueue.size():" + dataQueue.size());
			Object messageQueueService = SpringUtils.getObject("messageQueueService");
			// 循环取出队列中的所有数据
			MqMessage mqMessage;
			while ((mqMessage = dataQueue.poll()) != null) {
				mqMessage = mqMessage.toBuilder()
						.topic(daagProperties.getTopicName())
						.serviceName(daagProperties.getServiceName())
						.builder();
				if (!isOperationDelete(mqMessage)){
					List<Object> dbData = selectDataBaseDataByCondition(sqlSessionFactory,mqMessage);
					if(CollectionUtils.isEmpty(dbData)) {
					  log.warn("dbData is null");
		              continue;
					}
					mqMessage = mqMessage.toBuilder()
							.dbData(dbData)
							.builder();
				}
				if (messageQueueService instanceof MessageQueueService) {
					MessageQueueService messageQueueServiceBean = (MessageQueueService) messageQueueService;
					messageQueueServiceBean.sendMessage("rocketmq",mqMessage.getTable(),JSON.toJSONString(mqMessage,SerializerFeature.WriteMapNullValue));
				}
			}
			log.info("asyncQueryDbAndSendMsg() execution end: dataQueue.size():" + dataQueue.size());
			return Boolean.TRUE;
		}).whenComplete(new BiConsumer<Boolean, Throwable>() {
			@Override
			public void accept(Boolean result, Throwable throwable) {
				log.info("=====================异步操作end========================");
			}
		}).exceptionally(new Function<Throwable, Boolean>() {
			@Override
			public Boolean apply(Throwable throwable) {
				log.error("asyncQueryDbAndSendMsg() method error", throwable);
				return Boolean.FALSE;
			}
		});

	}
	public boolean isOperationDelete(MqMessage mqMessage) {
		Boolean delFlag = OperationTypeModel.DELETE.getModeLowerCase().equals(mqMessage.getOperType().toLowerCase());
		return delFlag;
	}
	/**
	 * 通过SqlSessionFactory查询业务数据库数据并拼装为JSON
	 * 
	 * @param sqlSessionFactory
	 * @param tableName
	 * @param whereCondition
	 * @return
	 */
	public List<Object> selectDataBaseDataByCondition(SqlSessionFactory sqlSessionFactory, MqMessage mqMessage) {
		List<Object> list = new ArrayList<Object>();
		String tableName = mqMessage.getTable();
	 	String whereCondition = mqMessage.getWhereQueryCondition();		
		SqlSession sqlSession = sqlSessionFactory.openSession();
		try {
			StringBuffer buffer = new StringBuffer();
			buffer.append(SqlConstants.SELECT).append(SqlConstants.FROM);
			if (StringUtils.isEmpty(tableName)) {
				return list;
			}
			buffer.append(" " + tableName + " ");
			if (!StringUtils.isEmpty(whereCondition)) {
				buffer.append(SqlConstants.WHERE + whereCondition.trim());
			}
			log.debug("select statement SQL: " + buffer.toString());
			sqlSession.getConfiguration().setDefaultStatementTimeout(60);
			Connection conn = sqlSession.getConnection();
            log.info(buffer.toString());
			PreparedStatement ps = conn.prepareStatement(buffer.toString());
			ResultSet rs = ps.executeQuery();
			ResultSetMetaData metaData = rs.getMetaData();
			int columnCount = metaData.getColumnCount();
			while (rs.next()) {
				JSONObject jsonObj = new JSONObject();
				// 遍历每一列
				for (int i = 1; i <= columnCount; i++) {
					String columnName = metaData.getColumnLabel(i);
					Object value = rs.getString(columnName);
					jsonObj.put(columnName, value);
				}
				list.add(jsonObj);
			}
		} catch (SQLException e) {
			e.printStackTrace();
			throw new DaagException(ErrorCodeEnum.SELECT_DBDATA_BYCONDITION_ERROR);
		} finally {
			if (sqlSession != null) {
				sqlSession.close();
			}
		}
		return list;
	}
}
